package com.k8.registry.config;


import com.k8.common.aop.ProxyFactory;
import com.k8.common.aop.SpecialBeanProxyFactory;
import com.k8.common.event.EventDeployer;
import com.k8.common.executor.ExecutorRepository;
import com.k8.common.factroy.BeanFactory;
import com.k8.common.factroy.loader.ExtensionDirector;
import com.k8.common.factroy.loader.ExtensionLoader;
import com.k8.registry.listener.NotifyListener;
import com.k8.rpc.Protocol;
import com.k8.rpc.cluster.Cluster;
import com.k8.rpc.cluster.ClusterInvoker;
import com.k8.rpc.directory.Directory;
import com.k8.rpc.directory.StaticDirectory;
import com.k8.rpc.invoke.Invoker;
import com.k8.rpc.invoke.invocation.Invocation;
import com.k8.rpc.result.Result;
import com.k8.rpc.url.ApiSignature;
import com.k8.rpc.url.URL;
import com.k8.rpc.util.RpcUtils;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.IllegalFormatCodePointException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import static com.k8.common.constants.CommonKeyConstants.EXPORTER_KEY;
import static com.k8.common.constants.CommonKeyConstants.PROXY_KEY;
import static com.k8.common.constants.CommonKeyConstants.RegistryKeyConstant.REGISTRY_KEY;
import static com.k8.common.constants.CommonKeyConstants.RpcKeyConstant.API_SIGNATURE_KEY;

/**
 * 扫描依赖，每个依赖都创建一个RefServiceConfig，通过subscribe订阅服务，并通过AbstractRegistry注册到本地，
 * 生成invoker还是通过refer，倒是需要给ConsumerInvoker指定cluster层实现，invoker类似于netty的handler
 * netty只负责传输层到应用的处理，通过各种handler
 * 多种invoker需要配合完成整个调用过程，此外定义好到底是invoker提供扩展还是Filter
 * <p>
 * consumer端需要对进行动态代理从而才能通过调用对象方法转而执行invoker
 */

public class ReferenceBase<T> extends AbstractReferenceConfig<T> {

    private final List<Invoker<T>> invokers = new CopyOnWriteArrayList<>();
    private final Protocol adaptiveProtocol;
    private final SpecialBeanProxyFactory specialBeanProxyFactory;
    private final ExecutorRepository executorRepository;
    private final EventDeployer eventDeployer;
    private final BeanFactory beanFactory;

    public ReferenceBase(BeanFactory beanFactory, List<URL> registryConfigUrls, URL systemConfigUrl) {
        super(systemConfigUrl, registryConfigUrls);
        ExtensionDirector extensionDirector = beanFactory.getExtensionDirector();
        this.adaptiveProtocol = extensionDirector.getExtensionLoader(Protocol.class).getAdaptiveExtension();
        this.specialBeanProxyFactory = beanFactory.getBean(SpecialBeanProxyFactory.class);
        if (this.specialBeanProxyFactory == null)
            throw new IllegalStateException("SpecialBeanProxyFactory is not register in beanFactory");
        ExtensionLoader<ExecutorRepository> executorLoader = extensionDirector.getExtensionLoader(ExecutorRepository.class);
        this.executorRepository = executorLoader.getDefaultExtension();
        this.eventDeployer = beanFactory.getBean(EventDeployer.class);
        if (this.eventDeployer == null) throw new IllegalStateException("EventDeployer is not register in beanFactory");
        this.beanFactory = beanFactory;
    }

    /**
     * 不同于ServiceBase，消费者端只适配一个协议，且只只指定协议名，具体的协议配置由该注册中心对该服务的协议的具体配置协商生成
     * 1.invokes主要针对多个注册中心，生成对应的clusterInvoker，在每个注册中心
     */

    public T createProxy() {
        NotifyListener notifyListener = new NotifyListener();
        eventDeployer.addListener(notifyListener);
        for (URL subscribeUrl : registryConfigUrls) {
            URL emptyProtocolUrl = new URL(systemConfigUrl);
            emptyProtocolUrl.setPath(RpcUtils.interfaceToPath(type, group, version));
            ApiSignature apiSignature = new ApiSignature(protocol, emptyProtocolUrl.getPath(), subscribeUrl.getPath());
            subscribeUrl.putAttribute(API_SIGNATURE_KEY, apiSignature);
            subscribeUrl.setParameter(EXPORTER_KEY, protocol);
            subscribeUrl.putAttribute(EXPORTER_KEY, emptyProtocolUrl);
            subscribeUrl.putAttribute(REGISTRY_KEY, subscribeUrl);
            //refer不仅是对invoker的创建，同样需要对mapping，configuration--创建该服务对应的protocolConfig，instance进行订阅
            //统一交给DynamicDirectory管理
            Invoker<T> invoker = adaptiveProtocol.refer(subscribeUrl, type);
            invokers.add(invoker);
        }
        Cluster<T> cluster = RpcUtils.getCluster(systemConfigUrl, beanFactory);
        StaticDirectory<T> staticDir = new StaticDirectory<>(invokers, type);
        //注册中心的LoadBalance基于staticDir
        T target = createInstance(staticDir, cluster, systemConfigUrl);
        return target;
    }

    /**
     * @param directory
     * @param cluster
     * @param systemConfigUrl
     * @return
     */
    private T createInstance(Directory<T> directory, Cluster cluster, URL systemConfigUrl) {
        ExtensionDirector extensionDirector = beanFactory.getExtensionDirector();
        ExtensionLoader<ProxyFactory> extensionLoader = extensionDirector.getExtensionLoader(ProxyFactory.class);
        ProxyFactory proxyFactory = extensionLoader.getExtension(systemConfigUrl.getParameter(PROXY_KEY));
        if (proxyFactory == null) {
            proxyFactory = extensionLoader.getDefaultExtension();
            if (proxyFactory == null) throw new IllegalStateException("proxyFactory is null");
        }
        ClusterInvoker<T> invoker = cluster.join(directory);
        ClusterInvoker<T> proxyInvoker = specialBeanProxyFactory.build(systemConfigUrl, invoker);
        InstanceToInvoker instanceToInvoker = new InstanceToInvoker(proxyInvoker);
        T reference = proxyFactory.createInterfaceProxy(type, instanceToInvoker);
        return reference;
    }

    private class InstanceToInvoker implements ProxyFactory.InterfaceProxyMethodInterceptor {
        private final Invoker<?> invoker;

        public InstanceToInvoker(Invoker<?> invoker) {
            this.invoker = invoker;
        }

        @Override
        public Object process(Object proxy, Method method, Object[] args) {
            Invocation invocation = RpcUtils.createConsumerInvocation(method, args, invoker.getInterface());
            try {
                Result<?> result = invoker.invoke(invocation);
                return result.getValue();
            } catch (Throwable e) {
                throw new RuntimeException(e);
            }
        }
    }
}
